#include "redis.hpp"

Redis::Redis() : _publish_context(nullptr), _subscribe_context(nullptr) {}

Redis::~Redis()
{
    if (_publish_context != nullptr)
    {
        redisFree(_publish_context);
    }
    if (_subscribe_context != nullptr)
    {
        redisFree(_subscribe_context);
    }
}

bool Redis::connect()
{
    // 负责发布的redis上下文连接
    _publish_context = redisConnect("127.0.0.1", 6379);
    if (_publish_context == nullptr)
    {
        cerr << "create publishContext error" << endl;
        return false;
    }
    _subscribe_context = redisConnect("127.0.0.1", 6379);
    if (_subscribe_context == nullptr)
    {
        cerr << "create subscribeContext error" << endl;
        return false;
    }
    // 因为订阅和发布都是阻塞式的，这里独立开辟一个线程做订阅工作
    thread t([&]()
             { observer_channel_message(); });
    t.detach();

    cout << "connect redis-server success" << endl;
    return true;
}
// 向channel中发布消息
bool Redis::publish(int channel, string message)
{
    redisReply *reply = (redisReply *)redisCommand(_publish_context, "PUBLISH %d %s", channel, message.c_str());
    if (reply == nullptr)
    {
        cerr << "publish message is failed" << endl;
        return false;
    }
    // 释放reply资源
    freeReplyObject(reply);
    return true;
}
// 订阅channel
bool Redis::subscribe(int channel)
{
    // 只发布订阅命令，不阻塞线程接收订阅的消息，这里接收消息交给了observer_channel_message()
    // 把函数中的redis命令缓存到本地
    if (REDIS_ERR == redisAppendCommand(_subscribe_context, "SUBSCRIBE %d", channel))
    {
        cerr << "redis subscribeCommand is failed" << endl;
        return false;
    }
    int done = 0;
    // 将发送缓冲区内的redis命令全部写入到redis服务器
    if (REDIS_ERR == redisBufferWrite(_subscribe_context, &done))
    {
        cerr << "redis subscribeCommand is failed" << endl;
        return false;
    }
    return true;
}
// 取消对channel的订阅
bool Redis::unsubscribe(int channel)
{
    if (REDIS_ERR == redisAppendCommand(this->_subscribe_context, "UNSUBSCRIBE %d", channel))
    {
        cerr << "unsubscribe command failed!" << endl;
        return false;
    }
    // redisBufferWrite可以循环发送缓冲区，直到缓冲区数据发送完毕（done被置为1）
    int done = 0;
    while (!done)
    {
        if (REDIS_ERR == redisBufferWrite(this->_subscribe_context, &done))
        {
            cerr << "unsubscribe command failed!" << endl;
            return false;
        }
    }
    return true;
}
// 接收订阅通道中的消息
void Redis::observer_channel_message()
{
    redisReply *reply = nullptr;
    while (REDIS_OK == redisGetReply(_subscribe_context,(void**)&reply))
    {

        if (reply != nullptr && reply->element[2] != nullptr && reply->element[2]->str != nullptr)
        {
            // 将消息发回业务层处理,第一个参数是接到消息的管道号
            _notify_message_handler(atoi(reply->element[1]->str), reply->element[2]->str);
        }
        freeReplyObject(reply);
    }
    cerr<<"-------redis observer stop--------"<<endl;
}
//设置redis收到订阅的消息后要进行的处理
void Redis::init_notify_handler(function<void(int, string)> fun)
{
    this->_notify_message_handler = fun;
}
